package pattern.socketmq;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.SerializationUtils;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import pattern.cache.InitCache;
import pattern.enums.mq.SubscribeTopicEnum;
import pattern.model.mq.SendMessageModel;
import pattern.mq.AbstractConsumerSubscribe;
import pattern.result.ResponseData;
import pattern.result.ResponseResultStatus;
import pattern.socketimpl.SocketSendProxy;
import pattern.utils.CollectionUtil;

import java.util.List;

/**
 * 对接其它系统发送消息给用户
 * @author Simon
 */
@Component
@Slf4j
@RequiredArgsConstructor(onConstructor_ = @Autowired)
public class SendMessageToUserMqConsumer extends AbstractConsumerSubscribe implements ApplicationRunner {

    @Override
    public void run(ApplicationArguments args) throws Exception {
        //初始化
        super.init();
    }

    private final SocketSendProxy socketSendProxy;
    private final InitCache initCache;

    @Override
    protected SubscribeTopicEnum getSubscribeTopicEnum() {
        return SubscribeTopicEnum.SEND_MESSAGE_TO_USER_PUSH;
    }

    @Override
    protected InitCache getInitCache() {
        return initCache;
    }

    @Override
    protected void consume(MessageExt msg, ConsumeConcurrentlyContext context) {
        SendMessageModel sendMessageModel = (SendMessageModel) SerializationUtils.deserialize(msg.getBody());
        if (CollectionUtil.isEmpty(sendMessageModel.getUserIds())) {
            log.error("无接收用户{}", sendMessageModel);
            return;
        }
        if (StringUtils.isBlank(sendMessageModel.getTopic())) {
            log.error("主题为空{}", sendMessageModel);
            return;
        }
        log.info("发送消息{}", JSON.toJSON(sendMessageModel));
        ResponseData<Object> data = new ResponseData<>(ResponseResultStatus.SUCCESS);
        data.setData(sendMessageModel.getMsgBody());
        socketSendProxy.toOneByUserId(sendMessageModel.getUserIds(), sendMessageModel.getTopic(), data);
    }

    @Override
    protected void exception(List<MessageExt> msg, ConsumeConcurrentlyContext context) {
        log.error("消费异常处理：{}",JSON.toJSONString(msg));
    }
}
